package com.nanrui.mutualinductor.communicate;

import android.util.Log;

import com.nanrui.mutualinductor.data.DKPacket;
import com.network.utils.LogUtils;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.Arrays;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.DelayQueue;
import java.util.logging.Level;
import java.util.logging.Logger;

public class PacketQueue<T extends DKPacket> {

    public static final String TAG = "PacketQueue_data";

    private static final long TIME_OUT = 1000L;
    private short frameNo = (short) 0;
    private InputStream inputStream;
    private boolean lock;
    private ConcurrentMap<Short, QueueElement> map = new ConcurrentHashMap();
    private OutputStream outputStream;
    private DelayQueue<QueueElement> queue = new DelayQueue();

    private ConnectedThread mConnectedThread;

    public PacketQueue(InputStream paramInputStream, OutputStream paramOutputStream) {
        this.inputStream = paramInputStream;
        this.outputStream = paramOutputStream;
        this.lock = true;

        //取消任何线程正在运行的连接
        if (mConnectedThread != null) {
            mConnectedThread.cancel();
            mConnectedThread = null;
        }

        // 启动线程管理连接和传输
        mConnectedThread = new ConnectedThread(paramInputStream, paramOutputStream);
        mConnectedThread.start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    while (lock) {
                        QueueElement localQueueElement = (QueueElement) PacketQueue.this.queue.take();
                        localQueueElement.setCount(localQueueElement.getCount() - 1);
                        if (localQueueElement.getDkPacket() == null) {

                            if (localQueueElement.getCount() > 0) {
//                                Logger localLogger = PacketQueue.LOG;
                                StringBuilder localStringBuilder = new StringBuilder();
                                localStringBuilder.append("重新发送：");
                                localStringBuilder.append(localQueueElement.getDkPacket().getFrameNo());
                                localStringBuilder.append(", 剩余重试次数");
                                localStringBuilder.append(localQueueElement.getCount());
//                                localLogger.info(localStringBuilder.toString());
                                PacketQueue.this.queue.add(localQueueElement);
                                PacketQueue.this.send(localQueueElement.getDkPacket());
                            } else {
//                                localQueueElement.getDkPacket().setMessageCode(DKMessageCode.PACKET_TIMEOUT);
                                localQueueElement.getCallback().packetReceived(localQueueElement.getDkPacket());
                                PacketQueue.this.map.remove(localQueueElement.getDkPacket().getFrameNo());
                            }
                        } else {
                            PacketQueue.this.map.remove(localQueueElement.getDkPacket().getFrameNo());
                        }
                    }
                } catch (InterruptedException localInterruptedException) {
//                    PacketQueue.LOG.error(DKMessageCode.QUEUE_INTERRUPTED.toString());
                }
            }
        }).start();
    }


    /* Error */
    private void send(DKPacket paramDKPacket) {

        //创建临时对象
        ConnectedThread r;
        // 同步副本的connectedthread
        synchronized (this) {
            r = mConnectedThread;
        }
        // 执行写同步
//        Log.e(TAG,"ConnectedThread write out:" + new String(out));
        r.write(paramDKPacket.pack((short) 1));
    }

    public void clear() {
        this.lock = false;
        this.map.clear();
        this.queue.clear();
    }

    public void put(DKPacket paramDKPacket, PacketCallback<T> paramPacketCallback) {
        try {
            QueueElement localQueueElement = new QueueElement(
                    paramDKPacket, paramPacketCallback, paramDKPacket.getCount(), 3 * 1000L);
            this.map.put(Short.valueOf(paramDKPacket.getFrameNo()), localQueueElement);
            this.queue.offer(localQueueElement);
            send(paramDKPacket);
            return;
        } finally {

        }
    }


    private ByteBuffer mRemainingBuf;
    private int headerLength = 8;

    /**
     * 本线在连接与远程设备。
     * 它处理所有传入和传出的传输。
     */
    private class ConnectedThread extends Thread {
        private final InputStream mmInStream;
        private final OutputStream mmOutStream;

        public ConnectedThread(InputStream inputStream, OutputStream outputStream) {
            Log.d(TAG, "create ConnectedThread");
            mmInStream = inputStream;
            mmOutStream = outputStream;
        }

        @Override
        public void run() {
            Log.i(TAG, "BEGIN mConnectedThread");
            int bytes;
            // 继续听InputStream同时连接
            while (true) {
                try {

                    OriginalData originalData = new OriginalData();
                    //读取头部
                    ByteBuffer headBuf = ByteBuffer.allocate(headerLength);
                    headBuf.order(ByteOrder.LITTLE_ENDIAN);

                    if (mRemainingBuf != null) {
                        mRemainingBuf.flip();
                        int length = Math.min(mRemainingBuf.remaining(), headerLength);
                        headBuf.put(mRemainingBuf.array(), 0, length);
                        if (length < headerLength) {
                            //there are no data left
                            mRemainingBuf = null;
                            readHeaderFromChannel(headBuf, headerLength - length);
                        } else {
                            mRemainingBuf.position(headerLength);
                        }
                    } else {
                        readHeaderFromChannel(headBuf, headBuf.capacity());
                    }
                    originalData.setHeadBytes(headBuf.array());
                    LogUtils.d("head", Arrays.toString(originalData.getHeadBytes()));

                    int[] headData = getBodyLength(originalData.getHeadBytes(), ByteOrder.LITTLE_ENDIAN);
                    int bodyLength = (short) (headData[0] & 0xFF) + 3;
                    short frameNo = (short) (headData[1] & 0xFF);
//                    if (SLog.isDebug()) {
//                        SLog.i("need read body length: " + bodyLength);
//                    }
                    if (bodyLength > 0) {
//                        if (bodyLength > mOkOptions.getMaxReadDataMB() * 1024 * 1024) {
//                            throw new ReadException("Need to follow the transmission protocol.\r\n" +
//                                    "Please check the client/server code.\r\n" +
//                                    "According to the packet header data in the transport protocol, the package length is " + bodyLength + " Bytes.\r\n" +
//                                    "You need check your <ReaderProtocol> definition");
//                        }
                        ByteBuffer byteBuffer = ByteBuffer.allocate(bodyLength);
                        byteBuffer.order(ByteOrder.LITTLE_ENDIAN);
                        if (mRemainingBuf != null) {
                            int bodyStartPosition = mRemainingBuf.position();
                            int length = Math.min(mRemainingBuf.remaining(), bodyLength);
                            byteBuffer.put(mRemainingBuf.array(), bodyStartPosition, length);
                            mRemainingBuf.position(bodyStartPosition + length);
                            if (length == bodyLength) {
                                if (mRemainingBuf.remaining() > 0) {//there are data left
                                    ByteBuffer temp = ByteBuffer.allocate(mRemainingBuf.remaining());
                                    temp.order(ByteOrder.LITTLE_ENDIAN);
                                    temp.put(mRemainingBuf.array(), mRemainingBuf.position(), mRemainingBuf.remaining());
                                    mRemainingBuf = temp;
                                } else {//there are no data left
                                    mRemainingBuf = null;
                                }
                                //cause this time data from remaining buffer not from channel.
//                                originalData.setBodyBytes(byteBuffer.array());
                                originalData.setBodyBytes(byteBuffer.array());
                                LogUtils.d("body", Arrays.toString(originalData.getBodyBytes()));
                                QueueElement element = map.get(frameNo);
                                DKPacket dkPacket = DKPacket.instance(byteMerger(originalData.getHeadBytes(), originalData.getBodyBytes()));
                                if (dkPacket != null && element.getCallback() != null) {
                                    element.getCallback().packetReceived(dkPacket);
                                    map.remove((short) frameNo);
                                }
//                                mStateSender.sendBroadcast(IOAction.ACTION_READ_COMPLETE, originalData);
                                return;
                            } else {//there are no data left in buffer and some data pieces in channel
                                mRemainingBuf = null;
                            }
                        }
                        readBodyFromChannel(byteBuffer);
                        originalData.setBodyBytes(byteBuffer.array());
                        LogUtils.d("body", Arrays.toString(originalData.getBodyBytes()));
//                        originalData.setBodyBytes(byteBuffer.array());
                    } else if (bodyLength == 0) {
//                        originalData.setBodyBytes(new byte[0]);
                        originalData.setBodyBytes(new byte[0]);
                        LogUtils.d("body", Arrays.toString(originalData.getBodyBytes()));
                        if (mRemainingBuf != null) {
                            //the body is empty so header remaining buf need set null
                            if (mRemainingBuf.hasRemaining()) {
                                ByteBuffer temp = ByteBuffer.allocate(mRemainingBuf.remaining());
                                temp.order(ByteOrder.LITTLE_ENDIAN);
                                temp.put(mRemainingBuf.array(), mRemainingBuf.position(), mRemainingBuf.remaining());
                                mRemainingBuf = temp;
                            } else {
                                mRemainingBuf = null;
                            }
                        }
                    } else if (bodyLength < 0) {
//                        throw new ReadException(
//                                "read body is wrong,this socket input stream is end of file read " + bodyLength + " ,that mean this socket is disconnected by server");
                    }

                    QueueElement element = map.get((short) frameNo);
                    DKPacket dkPacket = DKPacket.instance(byteMerger(originalData.getHeadBytes(), originalData.getBodyBytes()));
                    if (dkPacket != null && element != null && element.getCallback() != null) {
                        element.getCallback().packetReceived(dkPacket);
                        map.remove((short) frameNo);
                    }

//                    mStateSender.sendBroadcast(IOAction.ACTION_READ_COMPLETE, originalData);


//                    byte[] buffer = new byte[1];
//
//                    // 读取输入流
//                    int len = mmInStream.read(buffer);
//                    if (len == -1) {
//                        break;
//                    }


//                    LogUtils.d("sss", Arrays.toString(buffer));

                } catch (IOException e) {
                    Log.e(TAG, "disconnected", e);
//                    connectionLost();
                    break;
                }
            }
        }

        /**
         * 写输出的连接。
         *
         * @param buffer 这是一个字节流
         */
        public void write(byte[] buffer) {
            try {
                Log.e(TAG, "write buffer :" + new String(buffer));
                mmOutStream.write(buffer);
                mmOutStream.flush();
            } catch (IOException e) {
                Log.e(TAG, "Exception during write", e);
            }

        }

        public void cancel() {
//            try {
//                mmSocket.close();
//            } catch (IOException e) {
//                Log.e(TAG, "close() of connect socket failed", e);
//            }
        }
    }

    public static byte[] byteMerger(byte[] byte_1, byte[] byte_2) {
        byte[] byte_3 = new byte[byte_1.length + byte_2.length];
        System.arraycopy(byte_1, 0, byte_3, 0, byte_1.length);
        System.arraycopy(byte_2, 0, byte_3, byte_1.length, byte_2.length);
        return byte_3;
    }


    public int[] getBodyLength(byte[] header, ByteOrder byteOrder) {
        ByteBuffer bb = ByteBuffer.wrap(header);
        bb.order(byteOrder);
//        bb.putInt(2,4);
        int head = bb.getShort();
        int length = bb.getShort();
        int id = bb.get();
        int frameNo = bb.get();
        Logger.getGlobal().log(Level.WARNING, "guqian_socket " + length + "");
        return new int[]{length, frameNo};
    }

    private void readHeaderFromChannel(ByteBuffer headBuf, int readLength) throws IOException {
        for (int i = 0; i < readLength; i++) {
            byte[] bytes = new byte[1];
            int value = inputStream.read(bytes);
            if (value == -1) {

            }
            headBuf.put(bytes);
        }
    }

    private void readBodyFromChannel(ByteBuffer byteBuffer) throws IOException {
        while (byteBuffer.hasRemaining()) {
            try {
                byte[] bufArray = new byte[100];
                int len = inputStream.read(bufArray);
                if (len == -1) {
                    break;
                }
                int remaining = byteBuffer.remaining();
                if (len > remaining) {
                    byteBuffer.put(bufArray, 0, remaining);
                    mRemainingBuf = ByteBuffer.allocate(len - remaining);
                    mRemainingBuf.order(ByteOrder.LITTLE_ENDIAN);
                    mRemainingBuf.put(bufArray, remaining, len - remaining);
                } else {
                    byteBuffer.put(bufArray, 0, len);
                }
            } catch (Exception e) {
                throw e;
            }
        }
    }
}
